// 版权所有2016 etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

// 基于github。com/grpc-ecosystem/go-grpc-middleware/retry，但修改为支持etcd的write-at-one-retry语义所需的更
// 细粒度错误检查。

package clientv3

import (
	"context"
	"errors"
	"io"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

// unaryClientInterceptor返回一个新的重试一元客户端拦截器。
// 
// 拦截器的默认配置是根本不重试*。这种行为可以通过创建拦截器或调用时的选项（例如WithMax）（通过grpc.CallOptions）来改变。
func (c *Client) unaryClientInterceptor(optFuncs ...retryOption) grpc.UnaryClientInterceptor {
	intOpts := reuseOrNewWithCallOptions(defaultOptions, optFuncs)
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		ctx = withVersion(ctx)
		grpcOpts, retryOpts := filterCallOptions(opts)
		callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
		// 为简单起见短路，避免分配。
		if callOpts.max == 0 {
			return invoker(ctx, method, req, reply, cc, grpcOpts...)
		}
		var lastErr error
		for attempt := uint(0); attempt < callOpts.max; attempt++ {
			if err := waitRetryBackoff(ctx, attempt, callOpts); err != nil {
				return err
			}
			c.GetLogger().Debug(
				"retrying of unary invoker",
				zap.String("target", cc.Target()),
				zap.Uint("attempt", attempt),
			)
			lastErr = invoker(ctx, method, req, reply, cc, grpcOpts...)
			if lastErr == nil {
				return nil
			}
			c.GetLogger().Warn(
				"retrying of unary invoker failed",
				zap.String("target", cc.Target()),
				zap.Uint("attempt", attempt),
				zap.Error(lastErr),
			)
			if isContextError(lastErr) {
				if ctx.Err() != nil {
					// 这是上下文中的截止日期或取消日期。
					return lastErr
				}
				// 这是callCtx的截止日期或取消日期，在这种情况下请重试。
				continue
			}
			if c.shouldRefreshToken(lastErr, callOpts) {
				// 刷新前清除身份验证令牌。
				// 调用c.Auth。使用无效令牌进行身份验证将始终无法通过服务器端的身份验证检查，
				// 如果服务器尚未应用pr#12165的修补程序（https:
				// 和rpctypes.ErrInvalidAuthToken将递归调用c.getToken，直到系统资源耗尽。
				c.authTokenBundle.UpdateAuthToken("")

				gterr := c.getToken(ctx)
				if gterr != nil {
					c.GetLogger().Warn(
						"retrying of unary invoker failed to fetch new auth token",
						zap.String("target", cc.Target()),
						zap.Error(gterr),
					)
					return gterr // lastErr必须是无效的身份验证令牌
				}
				continue
			}
			if !isSafeRetry(c, lastErr, callOpts) {
				return lastErr
			}
		}
		return lastErr
	}
}

// streamClientInterceptor为服务器端流调用返回一个新的重试流客户端侦听器。
// 
// 拦截器的默认配置是完全不重试*。此行为可以通过选项（例如WithMax）在创建拦截器或呼叫时（通过grpc.CallOptions）更改.
// 
// 重试逻辑*仅适用于ServerStreams*，即1:n流，因为内部逻辑需要
// 来缓冲客户端发送的消息。如果在任何其他流（ClientStreams，
// BiDistream）上启用了重试功能，重试侦听器将使调用失败。
func (c *Client) streamClientInterceptor(optFuncs ...retryOption) grpc.StreamClientInterceptor {
	intOpts := reuseOrNewWithCallOptions(defaultOptions, optFuncs)
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		ctx = withVersion(ctx)
		// getToken自动
		// TODO（cfc4n）：保留此代码块，删除客户端中有关getToken的代码。在pr#12165合并后继续。
		if c.authTokenBundle != nil {
			// 等于c.用户名！=“和c.密码！”
			err := c.getToken(ctx)
			if err != nil && rpctypes.Error(err) != rpctypes.ErrAuthNotEnabled {
				c.GetLogger().Error("clientv3/retry_interceptor: getToken failed", zap.Error(err))
				return nil, err
			}
		}
		grpcOpts, retryOpts := filterCallOptions(opts)
		callOpts := reuseOrNewWithCallOptions(intOpts, retryOpts)
		// 为简单起见短路，避免分配。
		if callOpts.max == 0 {
			return streamer(ctx, desc, cc, method, grpcOpts...)
		}
		if desc.ClientStreams {
			return nil, status.Errorf(codes.Unimplemented, "clientv3/retry_interceptor: cannot retry on ClientStreams, set Disable()")
		}
		newStreamer, err := streamer(ctx, desc, cc, method, grpcOpts...)
		if err != nil {
			c.GetLogger().Error("streamer failed to create ClientStream", zap.Error(err))
			return nil, err // TODO（mwitkow）：可能应该重试拨号和传输错误？
		}
		retryingStreamer := &serverStreamingRetryingStream{
			client:       c,
			ClientStream: newStreamer,
			callOpts:     callOpts,
			ctx:          ctx,
			streamerCall: func(ctx context.Context) (grpc.ClientStream, error) {
				return streamer(ctx, desc, cc, method, grpcOpts...)
			},
		}
		return retryingStreamer, nil
	}
}

// shouldRefreshToken根据错误和调用选项检查是否需要刷新令牌，
// 并返回布尔值。
func (c *Client) shouldRefreshToken(err error, callOpts *options) bool {
	if rpctypes.Error(err) == rpctypes.ErrUserEmpty {
		// 当用户名、密码存在但服务器返回ErrUserEmpty 
		// 当客户端令牌以某种方式被清除时，这是可能的
		return c.authTokenBundle != nil // 等于c.Username！=''和&c.Password！=''
	}

	return callOpts.retryAuth &&
		(rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken || rpctypes.Error(err) == rpctypes.ErrAuthOldRevision)
}

// type serverStreamingRetryingStream是grpc.ClientStream的实现，充当基础调用的
// 代理。如果任何RecvMsg（）调用失败，它将尝试根据重试策略重新建立新的ClientStream。
type serverStreamingRetryingStream struct {
	grpc.ClientStream
	client        *Client
	bufferedSends []interface{} // 客户端可以发送的单个消息
	receivedGood  bool          // 表示之前的任何接收是否成功
	wasClosedSend bool          // 表示CloseSend已关闭
	ctx           context.Context
	callOpts      *options
	streamerCall  func(ctx context.Context) (grpc.ClientStream, error)
	mu            sync.RWMutex
}

func (s *serverStreamingRetryingStream) setStream(clientStream grpc.ClientStream) {
	s.mu.Lock()
	s.ClientStream = clientStream
	s.mu.Unlock()
}

func (s *serverStreamingRetryingStream) getStream() grpc.ClientStream {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.ClientStream
}

func (s *serverStreamingRetryingStream) SendMsg(m interface{}) error {
	s.mu.Lock()
	s.bufferedSends = append(s.bufferedSends, m)
	s.mu.Unlock()
	return s.getStream().SendMsg(m)
}

func (s *serverStreamingRetryingStream) CloseSend() error {
	s.mu.Lock()
	s.wasClosedSend = true
	s.mu.Unlock()
	return s.getStream().CloseSend()
}

func (s *serverStreamingRetryingStream) Header() (metadata.MD, error) {
	return s.getStream().Header()
}

func (s *serverStreamingRetryingStream) Trailer() metadata.MD {
	return s.getStream().Trailer()
}

func (s *serverStreamingRetryingStream) RecvMsg(m interface{}) error {
	attemptRetry, lastErr := s.receiveMsgAndIndicateRetry(m)
	if !attemptRetry {
		return lastErr // 成功还是硬失败
	}

	// 我们从尝试1开始，因为在正常的SendMsg（）上已经进行了第零次。
	for attempt := uint(1); attempt < s.callOpts.max; attempt++ {
		if err := waitRetryBackoff(s.ctx, attempt, s.callOpts); err != nil {
			return err
		}
		newStream, err := s.reestablishStreamAndResendBuffer(s.ctx)
		if err != nil {
			s.client.lg.Error("failed reestablishStreamAndResendBuffer", zap.Error(err))
			return err // TODO（mwitkow）：可能拨号和传输错误应可重试？
		}
		s.setStream(newStream)

		s.client.lg.Warn("retrying RecvMsg", zap.Error(lastErr))
		attemptRetry, lastErr = s.receiveMsgAndIndicateRetry(m)
		if !attemptRetry {
			return lastErr
		}
	}
	return lastErr
}

func (s *serverStreamingRetryingStream) receiveMsgAndIndicateRetry(m interface{}) (bool, error) {
	s.mu.RLock()
	wasGood := s.receivedGood
	s.mu.RUnlock()
	err := s.getStream().RecvMsg(m)
	if err == nil || err == io.EOF {
		s.mu.Lock()
		s.receivedGood = true
		s.mu.Unlock()
		return false, err
	} else if wasGood {
		// 流中以前的RecvMsg成功，重试逻辑不应干扰
		return false, err
	}
	if isContextError(err) {
		if s.ctx.Err() != nil {
			return false, err
		}
		// 这是callCtx截止日期或取消，在这种情况下请重试。
		return true, err
	}
	if s.client.shouldRefreshToken(err, s.callOpts) {
		// 清除身份验证令牌以避免调用getToken时失败
		s.client.authTokenBundle.UpdateAuthToken("")

		gterr := s.client.getToken(s.ctx)
		if gterr != nil {
			s.client.lg.Warn("retry failed to fetch new auth token", zap.Error(gterr))
			return false, err // 返回o为简单起见，原始错误
		}
		return true, err

	}
	return isSafeRetry(s.client, err, s.callOpts), err
}

func (s *serverStreamingRetryingStream) reestablishStreamAndResendBuffer(callCtx context.Context) (grpc.ClientStream, error) {
	s.mu.RLock()
	bufferedSends := s.bufferedSends
	s.mu.RUnlock()
	newStream, err := s.streamerCall(callCtx)
	if err != nil {
		return nil, err
	}
	for _, msg := range bufferedSends {
		if err := newStream.SendMsg(msg); err != nil {
			return nil, err
		}
	}
	if err := newStream.CloseSend(); err != nil {
		return nil, err
	}
	return newStream, nil
}

func waitRetryBackoff(ctx context.Context, attempt uint, callOpts *options) error {
	waitTime := time.Duration(0)
	if attempt > 0 {
		waitTime = callOpts.backoffFunc(attempt)
	}
	if waitTime > 0 {
		timer := time.NewTimer(waitTime)
		select {
		case <-ctx.Done():
			timer.Stop()
			return contextErrToGrpcErr(ctx.Err())
		case <-timer.C:
		}
	}
	return nil
}

// isSafeRetry返回“true”，如果请求在给定的错误下可以安全重试。
func isSafeRetry(c *Client, err error, callOpts *options) bool {
	if isContextError(err) {
		return false
	}

	// 当学习者拒绝它应该不提供的RPC时，情况是来自服务器
	// 透视图不可重试。
	// 但出于向后兼容性的原因，我们需要支持
	// 客户提供的混合学习者（尚未投票者）和有
	// 期望在下一次尝试中选择投票者。
	// TODO:理想情况下，客户应该“知道”哪个端点代表：高概率的领导者/选民/学习者。
	if errors.Is(err, rpctypes.ErrGPRCNotSupportedForLearner) && len(c.Endpoints()) > 1 {
		return true
	}

	switch callOpts.retryPolicy {
	case repeatable:
		return isSafeRetryImmutableRPC(err)
	case nonRepeatable:
		return isSafeRetryMutableRPC(err)
	default:
		c.lg.Warn("unrecognized retry policy", zap.String("retryPolicy", callOpts.retryPolicy.String()))
		return false
	}
}

func isContextError(err error) bool {
	return status.Code(err) == codes.DeadlineExceeded || status.Code(err) == codes.Canceled
}

func contextErrToGrpcErr(err error) error {
	switch err {
	case context.DeadlineExceeded:
		return status.Errorf(codes.DeadlineExceeded, err.Error())
	case context.Canceled:
		return status.Errorf(codes.Canceled, err.Error())
	default:
		return status.Errorf(codes.Unknown, err.Error())
	}
}

var (
	defaultOptions = &options{
		retryPolicy: nonRepeatable,
		max:         0, // 禁用
		backoffFunc: backoffLinearWithJitter(50*time.Millisecond /*jitter*/, 0.10),
		retryAuth:   true,
	}
)

// backoffFunc表示一系列函数，用于控制调用重试之间的退避持续时间。
// 
// 调用它们时使用了尝试的标识符，并且应该返回系统客户端应该等待的时间
// 延迟。如果返回的时间长于`上下文。上下文请求的截止日期
// 请求的截止日期优先，在进行下一次迭代之前，等待将被中断。
type backoffFunc func(attempt uint) time.Duration

// withRetryPolicy设置此调用的重试策略。
func withRetryPolicy(rp retryPolicy) retryOption {
	return retryOption{applyFunc: func(o *options) {
		o.retryPolicy = rp
	}}
}

// withMax设置此调用或此侦听器的最大重试次数。
func withMax(maxRetries uint) retryOption {
	return retryOption{applyFunc: func(o *options) {
		o.max = maxRetries
	}}
}

// WithBackoff设置用于控制重试间隔时间的'BackoffFunc'。
func withBackoff(bf backoffFunc) retryOption {
	return retryOption{applyFunc: func(o *options) {
		o.backoffFunc = bf
	}}
}

type options struct {
	retryPolicy retryPolicy
	max         uint
	backoffFunc backoffFunc
	retryAuth   bool
}

// retryOption是grpc。clientv3的重试侦听器本地的CallOption。
type retryOption struct {
	grpc.EmptyCallOption // 确保实现private after（）和before（）字段，这样我们就不会惊慌失措。
	applyFunc            func(opt *options)
}

func reuseOrNewWithCallOptions(opt *options, retryOptions []retryOption) *options {
	if len(retryOptions) == 0 {
		return opt
	}
	optCopy := &options{}
	*optCopy = *opt
	for _, f := range retryOptions {
		f.applyFunc(optCopy)
	}
	return optCopy
}

func filterCallOptions(callOptions []grpc.CallOption) (grpcOptions []grpc.CallOption, retryOptions []retryOption) {
	for _, opt := range callOptions {
		if co, ok := opt.(retryOption); ok {
			retryOptions = append(retryOptions, co)
		} else {
			grpcOptions = append(grpcOptions, opt)
		}
	}
	return grpcOptions, retryOptions
}

// 带抖动的回退线等待设定的时间段，允许抖动（分数调整）。
// 
// 例如，waitBetween=1s和jitter=0.10可以生成900ms到1100ms之间的等待。
func backoffLinearWithJitter(waitBetween time.Duration, jitterFraction float64) backoffFunc {
	return func(attempt uint) time.Duration {
		return jitterUp(waitBetween, jitterFraction)
	}
}
